package com.kyd.push.mq.config;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.kyd.push.mq.properties.MQProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * CustomerConfig
 *
 * @author kejw on 2018/9/27
 */
@Slf4j
//@Configuration
public class MQCustomerConfig {

	@Autowired
	private MQProperties mqProperties;

	@Autowired
	private Map<String, MessageListener> listenerMap;

	@Bean(name = "customer", initMethod = "start", destroyMethod = "shutdown")
	public ConsumerBean customer() {
		log.info("[MQ]初始化并启动消费者......");

		Properties properties = new Properties();
		// 您在 MQ 控制台创建的 Producer ID
		properties.put(PropertyKeyConst.ConsumerId, mqProperties.getAliyun().getCustomer().getCustomerId());
		// 鉴权用 AccessKey，在阿里云服务器管理控制台创建
		properties.put(PropertyKeyConst.AccessKey, mqProperties.getAliyun().getCustomer().getAccessKey());
		// 鉴权用 SecretKey，在阿里云服务器管理控制台创建
		properties.put(PropertyKeyConst.SecretKey, mqProperties.getAliyun().getCustomer().getSecretKey());
		// 设置消费者线程数
		properties.setProperty(PropertyKeyConst.ConsumeThreadNums, mqProperties.getAliyun().getCustomer().getConsumeThreadNums());
		// 设置 TCP 接入域名（此处以公共云的公网接入为例）
		properties.put(PropertyKeyConst.ONSAddr, mqProperties.getAliyun().getCustomer().getOnsAddr());

		// 注册消息监听器
		Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();
		if (listenerMap != null) {
			for (Map.Entry<String, MessageListener> entry : listenerMap.entrySet()) {
				Subscription subscription = new Subscription();
				subscription.setTopic(entry.getKey()); // 注意：监听器需要命名方式，Topic名称与监听器名称保存一致
				subscription.setExpression("*");
				subscriptionTable.put(subscription, entry.getValue());
			}
		}

		ConsumerBean consumer = new ConsumerBean();
		consumer.setProperties(properties);
		consumer.setSubscriptionTable(subscriptionTable);
		return consumer;
	}

}
